package com.rrd.dw.mr.gzip.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.tongdun.TongdunRiskDetail;
import com.ucredit.babel.common.model.tongdun.TongdunRiskReport;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;

/**
 * 同盾风险报告-贷前审核借款事件
 *
 * @author xj
 */

public class TongdunRiskReportParse extends ParseBase {
    private Text val = new Text();
    private Text key = new Text();

    /**
     * 同盾风险报告-贷前审核借款事件
     *
     * @param model
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        TongdunRiskReport obj = (TongdunRiskReport) model;
        TongdunRiskDetail data = obj.getTongdunRiskDetail();
        JSONObject risk = data.getRisk() == null ? new JSONObject() : JSON.parseObject(data.getRisk());
        StringBuilder sb = getBaseInfo(obj, successful);
        //是否成功
        sb.append(risk.getString("success") == null ? "" : risk.getString("success")).append(CRT001);
        //若success为false，此处显示错误码
        sb.append(risk.getString("reason_code") == null ? "" : risk.getString("reason_code")).append(CRT001);
        //请求id
        sb.append(risk.getString("seq_id") == null ? "" : risk.getString("seq_id")).append(CRT001);
        //风险评估结果
        sb.append(risk.getString("final_decision") == null ?
                risk.getString("finalDecision") == null ? "" : risk.getString("finalDecision")
                : risk.getString("final_decision")
        ).append(CRT001);
        //风险系数
        sb.append(risk.getString("final_score") == null ?
                risk.getString("finalScore") == null ? "" : risk.getString("finalScore")
                : risk.getString("final_score")
        ).append(CRT001);

        sb.append(risk.getString("idNumberMobileScore") == null ? "" : risk.getString("idNumberMobileScore")).append(CRT001);
        //触发的策略集名称
        sb.append(risk.getString("policy_set_name") == null ?
                risk.getString("policySetName") == null ? "" : risk.getString("policySetName")
                : risk.getString("policy_set_name")
        ).append(CRT001);
        //命中的风险类型的标识及风险结果的集合
        sb.append(risk.getString("risk_type") == null ?
                risk.getString("riskType") == null ? "" : risk.getString("riskType")
                : risk.getString("risk_type")
        ).append(CRT001);
        //该接口内部处理耗时（ms)
        sb.append(risk.getString("spend_time") == null ?
                risk.getString("spendTime") == null ? "" : risk.getString("spendTime")
                : risk.getString("spend_time")
        ).append(CRT001);
        //包含每个策略以及策略的命中规则详见后面policy_set字段 JSON
        sb.append(risk.getString("policy_set") == null ?
                risk.getString("policySet") == null ? "" : risk.getString("policySet")
                : risk.getString("policy_set")).append(CRT001);
        //命中规则 JSON
        sb.append(risk.getString("hit_rules") == null ?
                risk.getString("hitRules") == null ? "" : risk.getString("hitRules")
                : risk.getString("hit_rules")).append(CRT001);
        setFileNum(5);
        key.set(getKey());
        val.set(sb.toString().replace("\n", " ").replace("\r", " "));
        context.write(key, val);
        return true;
    }

    @Override
    public String getSubType() {
        return ".risk";
    }

    @Override
    public String getTableName() {
        return "kafka_babel_tongdun_risklend_inc";
    }

    @Override
    public String getType() {
        return RecordType.TONGDUN_RISK_REPORT.toString();
    }

    @Override
    public String getComment() {
        return "同盾风险报告-贷前审核借款事件";
    }
}
